package com.stone.bigdata.flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
import org.apache.flink.types.Row;

import java.util.Map;

/**
 * @author fujianjian
 * @date 2023/8/9 16:44
 **/
public class MysqlInputFormatV1 {

    public static void main(String[] args) throws Exception {
        // step1: 获取环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // step2: 构建source
        DataSource<Row> ihrDatainput = env.createInput(JDBCInputFormat.buildJDBCInputFormat()
                .setDrivername("com.mysql.cj.jdbc.Driver")
                .setDBUrl("jdbc:mysql://10.89.2.119:3306/ihr_core?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8")
                .setUsername("ihr_read_gly").setPassword("Ihr@zb$2021")
                // 设置需要执行的SQL语句
                .setQuery("select ee.employee_id employeeId, ee.employee_code employeeCode, ee.employee_name employeeName, " +
                        "ee.id_number idNumber, ee.mobile mobile from t_emp_employee ee left join t_org_unit ou on ee.unit_id = ou.unit_id " +
                        "where ou.`path`  like '80000001_80030015_82010921_80037278_%'")
                // 设置查询列的返回类型
                .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)).setFetchSize(1000)
                .finish());

        // step3: transformation 转换
        DataSet<Map<String, String>> ihrEmpDataset = ihrDatainput.map(new MapFunction<Row, Map<String, String>>() {
            @Override
            public Map<String, String> map(Row row) throws Exception {
                Map<String, String> retMap = Maps.newHashMap();
                retMap.put("empId", String.valueOf(row.getField(0)));
                retMap.put("empCode", (String) row.getField(1));
                retMap.put("empName", (String) row.getField(2));
                retMap.put("empIdnum", (String) row.getField(3));
                retMap.put("empMobile", (String) row.getField(4));
                return retMap;
            }
        });

        // step4: Slink 打印到控制台
        ihrDatainput.print();
        // ihrEmpDataset.print();
        // long count = ihrEmpDataset.count();
        // System.out.println(count);

        // env.execute("ihr-emp-jdbc-input-format");
    }
}
